package com.sduept.component.message.core;

import com.sduept.component.message.core.config.DynamicTopicsConfig;
import com.sduept.component.message.model.TopicSubscriptionInfo;
import com.sduept.core.dao.SuperDAO;
import com.sduept.message.entity.AlarmMessage;
import com.sduept.message.entity.Message;
import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;


@Service
@Slf4j
public class UserAlarmMessageService {
    public static final String split_remark = ",";
    @Autowired
    private SuperDAO dao;
    @Autowired
    DynamicTopicsConfig dynamicTopicsConfig;
    /**
     * key 主题名称
     * value  该主题用户订阅信息
     */
    private Map<String, TopicSubscriptionInfo> topicInfoMap = new ConcurrentHashMap<String, TopicSubscriptionInfo>();

    @PostConstruct
    public void init() {
        List<TopicSubscriptionInfo> slist = dao.findAll(TopicSubscriptionInfo.class);
        for (TopicSubscriptionInfo info : slist) {
            topicInfoMap.put(info.getTopic(), info);
        }
//        subscribeMessage(AlarmMessage.class);
    }

    public Map<String, TopicSubscriptionInfo> getTopicInfoMap() {
        return topicInfoMap;
    }

    /**
     * 更改订阅信息
     *
     * @param info
     */
    public void updateTopicSubscriptionInfo(TopicSubscriptionInfo info) {
        if (StringUtils.isEmpty(info.getSubscriber())) {
            topicInfoMap.remove(info.getTopic());
            dao.delete(info.getId(), TopicSubscriptionInfo.class);
        } else {
            topicInfoMap.put(info.getTopic(), info);
            dao.createOrUpdate(info, TopicSubscriptionInfo.class);
        }
    }

    /**
     * 动态消息订阅
     * @param topic
     * @param record
     */
//    @KafkaListener(topics = {"#{'${DynamicTopics}'.split(',')}"})
    public void onMessage(@Header(KafkaHeaders.RECEIVED_TOPIC) String topic, ConsumerRecord<?, ?> record) {
        log.info("接收到消息：" + topic);
        TopicSubscriptionInfo info = topicInfoMap.get(topic);
        if (info != null && !StringUtils.isEmpty(info.getSubscriber())) {
            String[] subscribers = info.getSubscriber().split(split_remark);
            int updateCount = 0;
            for (String subscriber : subscribers) {
                try {
                    Message msg = new AlarmMessage();
                    Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                    if (kafkaMessage.isPresent()) {
                        log.info(record.value().toString());
                        JSONObject jbean = JSONObject.fromObject(kafkaMessage.get());
                        msg.setCreateTime(new Date());
                        if (!StringUtils.isEmpty(jbean.get("source").toString()))
                            msg.setSource(jbean.get("source").toString());
                        if (!StringUtils.isEmpty(jbean.get("timeliness").toString()))
                            msg.setTimeliness(Long.parseLong("" + jbean.get("timeliness").toString()));
                        msg.setTarget(subscriber);
                        msg.setMsgType(topic);
                        msg.setContent(jbean.get("content").toString());
                        dao.createOrUpdate(msg, Message.class);
                        updateCount++;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            log.info("更新消息告警："+updateCount);
        }
    }

}
